package cn.lzd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by lzd on 2018/12/20.
  */
object TransformationDemo {
  def main(args: Array[String]): Unit = {
    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("local-demo1"))

    demo8(sc)
  }

  /**
    * cartesian 笛卡儿积
    *
    * @param sc
    */
  def demo9(sc: SparkContext): Unit = {
    val studentList = Array(Tuple2(1, "leo"), Tuple2(2, "jack"), Tuple2(1, "tom"));
    val scoreList = Array(Tuple2(1, 100), Tuple2(2, 90), Tuple2(2, 60), Tuple2(2, 70), Tuple2(3, 60));
    val rdd1 = sc.parallelize(studentList, 1)
    val rdd2 = sc.parallelize(scoreList, 1)
    val rdd3 = rdd1.cartesian(rdd2)
    rdd3.foreach(rdd => {
      println(rdd._1._1 + "--" + rdd._1._2)
      println(rdd._2._1 + "--" + rdd._2._2)
      println("**" * 20)
    })
  }


  /**
    * join
    * 在类型为（K,V)和（K,W)类型的数据集上调用，返回一个（K,(V,W))对，每个key中的所有元素都在一起的数据集
    *
    * @param sc
    */
  def demo7(sc: SparkContext): Unit = {
    val studentList = Array(Tuple2(1, "leo"), Tuple2(2, "jack"), Tuple2(4, "tom"));
    val scoreList = Array(Tuple2(1, 100), Tuple2(2, 90), Tuple2(3, 60));
    val rdd1 = sc.parallelize(studentList, 1)
    val rdd2 = sc.parallelize(scoreList, 1)
    val rdd3: RDD[(Int, (String, Int))] = rdd1.join(rdd2)

    rdd3.foreach(rdd => {
      println(f"id=${rdd._1}")
      println(f"name=${rdd._2._1}")
      println(f"score=${rdd._2._2}")
    })

  }

  /**
    * reduceByKey
    * 在一个（K，V)对的数据集上使用，返回一个（K，V）对的数据集，key相同的值，都被使用指定的reduce函数聚合到一起。
    * 和groupbykey类似，任务的个数是可以通过第二个可选参数来配置的。在实际开发中，能使reduceByKey实现的就不用groupByKey
    *
    * @param sc
    */
  def demo6(sc: SparkContext): Unit = {
    val lst = Array(Tuple2("hello", 1), Tuple2("hello", 2), Tuple2("world", 3))
    val rdd1 = sc.parallelize(lst, 1)
    val reduceByKey: RDD[(String, Int)] = rdd1.reduceByKey(_ + _)
    reduceByKey.foreach(rdd => println(rdd._1 + "\t" + rdd._2))
  }

  /**
    * groupByKey 结果：ArrayBuffer((hello,CompactBuffer(1, 2)), (world,CompactBuffer(3)))
    * 在一个由（K,V）对组成的数据集上调用，返回一个（K，Seq[V])对的数据集。注意：默认情况下，使用8个并行任务进行分组，你可以传入numTask可选参数，根据数据量设置不同数目的Task
    *
    * @param sc
    */
  def demo5(sc: SparkContext): Unit = {
    val lst = Seq(Tuple2("hello", 1), Tuple2("hello", 2), Tuple2("world", 3))
    val rdd1 = sc.parallelize(lst, 1)
    val rdd2 = rdd1.groupByKey()
    rdd2.foreach(rdd => {
      println(rdd._1)
      rdd._2.foreach(println(_))
    })
    println(rdd2.collect().toBuffer)
  }

  /**
    * cogroup 结果：ArrayBuffer((hello,(CompactBuffer(1, 2),CompactBuffer(11, 22))), (world,(CompactBuffer(3),CompactBuffer(3))))
    * @param sc
    */
  def demo5_1(sc: SparkContext): Unit = {
    val lst1 = Seq(Tuple2("hello", 1), Tuple2("hello", 2), Tuple2("world", 3))
    val lst2 = Seq(Tuple2("hello", 11), Tuple2("hello", 22), Tuple2("world", 3))
    val rdd1 = sc.parallelize(lst1, 1)
    val rdd2 = sc.parallelize(lst2, 1)

    //    val rdd1 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
    //    val rdd2 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("shuke", 2)))

    val cogroup = rdd1.cogroup(rdd2)

    cogroup.foreach((x) => {
      println(x._1)
    })
    println(cogroup.collect().toBuffer)
  }

  /**
    * groupWith
    * 在类型为（K,V)和(K,W)类型的数据集上调用，返回一个数据集，组成元素为（K, Seq[V], Seq[W]) Tuples。这个操作在其它框架，称为CoGroup
    */
  //TODO 没搞懂
  def demo8(sc: SparkContext): Unit = {
    val studentList = Array(Tuple2(1, "leo"),Tuple2(1, 111), Tuple2(2, "jack"), Tuple2(1, "tom"));
    val scoreList = Array(Tuple2(1, 100), Tuple2(2, 90), Tuple2(2, 60), Tuple2(2, 70), Tuple2(3, 60));
    val rdd1 = sc.parallelize(studentList, 1)
    val rdd2 = sc.parallelize(scoreList, 1)
    val rdd3 = rdd1.groupWith(rdd2)
    rdd3.foreach(add => {
      println(add._1)

      println(add._2._1.foreach(println(_)))
      println(add._2._2.foreach(println(_)))
      println("**" * 20)
    })

    println(rdd3.collect().toBuffer)
  }

  /**
    * union 求并集
    * 返回一个新的数据集，由原数据集和参数联合而成
    *
    * @param sc
    */
  def demo4(sc: SparkContext): Unit = {
    val lst = List(1, 2, 3, 4)
    val lst2 = Array(5, 6, 7)

    val rdd1 = sc.parallelize(lst)
    val rdd2 = sc.parallelize(lst2)
    val rdd3 = rdd1.union(rdd2)
    rdd3.foreach(println(_))
  }

  /**
    * intersection 求交集
    * 返回一个新的数据集，由原数据集和参数联合而成
    *
    * @param sc
    */
  def demo4_1(sc: SparkContext): Unit = {
    val lst = List(1, 2, 3, 4)
    val lst2 = Array(5, 3, 7)

    val rdd1 = sc.parallelize(lst)
    val rdd2 = sc.parallelize(lst2)
    val rdd3 = rdd1.intersection(rdd2)
    rdd3.foreach(println(_))
  }


  /**
    * flatMap
    * 类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）
    *
    * @param sc
    */
  def demo3(sc: SparkContext): Unit = {
    val lst = List("hello world", "my name", "study spark");
    val rdd1 = sc.parallelize(lst, 1)
    val rdd2 = rdd1.flatMap(_.split(" "))
    rdd2.foreach(println(_))
  }

  /**
    * filter
    * 返回一个新的数据集，由经过func函数后返回值为true的原元素组成
    *
    * @param sc
    */
  def demo2(sc: SparkContext): Unit = {
    val lst = List(2, 3, 4, 5, 6)
    val rdd1 = sc.parallelize(lst)
    val rdd2 = rdd1.filter(_ % 2 == 0)
    rdd2.foreach(println(_))
  }

  /**
    * map
    * 返回一个新的分布式数据集，由每个原元素经过func函数转换后组成
    *
    * @param sc
    */
  def demo1(sc: SparkContext): Unit = {
    val seq = Seq(2, 4, 5, 6, 8, 9)
    val rdd1 = sc.parallelize(seq, 1)
    val rdd2 = rdd1.map(_ * 2)
    rdd2.foreach(println(_))
  }
}
